#![allow(non_camel_case_types)]

pub mod common;
pub mod config;
pub mod controller;
pub mod db;
pub mod model;
pub mod route;


mod protoHeader;
mod Conn;
mod ProtoId;
mod receiveProtoHandler;
mod protos;


use log_util;
use tracing::Level;
use tracing_subscriber::FmtSubscriber;
use std::sync::{RwLock};
use tokio::sync::Notify;
use tokio::sync::mpsc::{channel, Sender, Receiver};
use std::collections::HashMap;
use std::{env, thread};
use log_util::LogType;
use tokio::io::{ReadHalf, WriteHalf, AsyncWriteExt, AsyncReadExt};
use tokio::net::TcpStream;
use tokio::task::JoinHandle;
use lazy_static;
use once_cell::sync::OnceCell;
use crate::protoHeader::ProtoHeader;
use tokio::time::Duration;
lazy_static::lazy_static! {
    pub static ref notify: Notify = Notify::new();
    pub static  ref rspMap: RwLock<HashMap<i32/*serialNO*/, Box<rspData>>>= RwLock::new(HashMap::new());
}

pub static PROTO_SENDER: OnceCell<Conn::Proto_sender> = OnceCell::new();

pub struct rspData {
    pub header: protoHeader::ProtoHeader,
    pub body_buffer: Vec<u8>,
    pub create_time: i64,
}

#[tokio::main]
async fn main() {
    let addr = get_args();
    let mut socket = tokio::net::TcpStream::connect(addr.as_str()).await
        .unwrap_or_else(
            |error|
                {
                    log_util::write_log_print(true, LogType::Error, error.to_string(), &None);
                    panic!("connect to server err:{}", error.to_string());
                }
        );
    let (mut read_half, mut write_half) = tokio::io::split(socket);

    let (sender, mut receiver): (Sender<Vec<u8>>, Receiver<Vec<u8>>) = channel(100);
    //
    let proto_sender = Conn::Proto_sender::new(sender);
    PROTO_SENDER.set(proto_sender);
    ProtoHeader::init();

    let send_proto_to_server = send_proto_to_server(write_half, receiver);

    let (sender, mut receiver) = channel(5);
    let receive_proto_from_srv = receive_proto_from_srv(read_half, sender);
    request_time_out_handle();

    thread_web();

    receiver.recv().await;
}

//deal with request time out to keep rsp_map clean
fn request_time_out_handle() {
    tokio::spawn(
        async {
            let dur = 10;
            let mill_second_span = (dur * 1000) as i64;
            loop {
                let mut interval1 = tokio::time::interval(std::time::Duration::from_secs(dur));
                interval1.tick().await;

                let now = time_util::now_unix();
                let guard = rspMap.read().unwrap();
                let mut need_remove = Vec::new();
                for v in guard.values() {
                    if now - v.create_time > mill_second_span {
                        need_remove.push(v.header.nSerialNo);
                    }
                }
                drop(guard);
                let mut write_rspMap = rspMap.write().unwrap();
                for k in need_remove{
                    write_rspMap.remove(&k);
                    // println!("rsp map time out remove :{:?}",k);
                }
            }
        }
    );
}

fn receive_proto_from_srv(mut readHalf: ReadHalf<TcpStream>, sender_to_main: tokio::sync::mpsc::Sender<()>) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut client_buffer = [0u8; 2048];
        // let mut readHalf = readHalf;
        println!("receiv start");
        loop {
            // println!("receiv loop start");
            let sender_clone = sender_to_main.clone();

            let readCount = readHalf.read(&mut client_buffer).await;
            tokio::spawn(async move {
                match readCount {
                    Ok(n) => {
                        if n == 0 {
                            println!("receive from server: 0");
                            // program.exit(0);
                        } else {
                            // println!("begin analyse:{:?}", readCount);
                            match receiveProtoHandler::analyse_header(&client_buffer, n) {
                                Ok(header) => {
                                    let mut rsp_map = rspMap.write().unwrap();
                                    let body_buffer = Vec::from(&client_buffer[protoHeader::HEADER_SIZE..n]);
                                    // println!("receive from server bodylen:{:?}", header.nBodyLen);
                                    // println!("rsp_map insert serilaNO:{:?}", header.nSerialNo);
                                    rsp_map.insert(header.nSerialNo, Box::new(rspData {
                                        header,
                                        body_buffer,
                                        create_time: time_util::now_unix(),
                                    }));

                                    notify.notify_waiters();
                                }
                                Err(e) => {
                                    println!("receive from server err: {:?}", e);

                                    log_util::local_error!("analyseReceiveMsg from srv err:{:?}",e);
                                }
                            }
                        }
                    }
                    Err(error) => {
                        let msg = format!("receive from srv err:{:?}", error.to_string());
                        log_util::local_error!(msg);
                        // println!("send msg to main to exit");

                        //exit
                        sender_clone.send(()).await;
                    }
                };
            });
        };
    })
}

fn send_proto_to_server(mut write_half: WriteHalf<TcpStream>, mut receiver: Receiver<Vec<u8>>) -> JoinHandle<()> {
    tokio::spawn(async move {
        println!("send start");
        while let Some(b) = receiver.recv().await {
            let bytesArr = b.as_slice();
            if bytesArr.len() == 0 {
                continue;
            }

            write_half.write_all(bytesArr).await.or_else(
                |err| {
                    log_util::local_error!("send proto to srv err:{}",err.to_string());
                    Err(err)
                }
            );
            write_half.flush().await.or_else(
                |err| {
                    log_util::local_error!("send proto to srv flush err:{}",err.to_string());
                    Err(err)
                });
        }

        println!("send end");
    })
}

async fn connect_to_srv() -> (ReadHalf<TcpStream>, WriteHalf<TcpStream>) {
    let addr = get_args();
    let socket = tokio::net::TcpStream::connect(addr.as_str()).await
        .unwrap_or_else(
            |error|
                {
                    log_util::write_log_print(true, LogType::Error, error.to_string(), &None);
                    panic!("connect to server err:{}", error.to_string());
                }
        );
    tokio::io::split(socket)
}

type Port = u16;

fn get_args() -> String {
    let mut args = env::args();
    args.next().expect("");
    args.next().expect("2ed arg need ip addr")
}

fn thread_web() {
    tokio::spawn(async {
        config::init().await;

        log_util::write_log_print(
            true,
            log_util::LogType::Info,
            format!("init db connection"),
            &None,
        ).await;

        // db::init().await;

        log_util::write_log_print(
            true,
            log_util::LogType::Info,
            "start web service".to_string(),
            &None,
        ).await;

        // a builder for `FmtSubscriber`.
        let subscriber = FmtSubscriber::builder()
            // all spans/events with a level higher than TRACE (e.g, debug, info, warn, etc.)
            // will be written to stdout.
            .with_max_level(Level::INFO)
            // completes the builder.
            .finish();
        tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");

        route::init().await;
    });
}
